/**
 * @author: yasen
 * @time: 2022/10/10 16:29
 * @email: yasen@yasenstudio.com
 */
package com.yasenstudio.flow.tools;

import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Properties;

public class SimpleKafkaConsumerBuilder {
    private String bootstrap;
    private String groupId;
    private String enableAutoCommit;
    private String autoCommitIntervalMs;
    private String autoOffsetReset;
    private String sessionTimeoutMs;
    private String keyDeserializer;
    private String valueDeserializer;

    public SimpleKafkaConsumerBuilder(String bootstrap) {
        this.bootstrap = bootstrap;
        this.groupId = "default_group";
        this.enableAutoCommit = "true";
        this.autoCommitIntervalMs = "1000";
        this.autoOffsetReset = "earliest";
        this.sessionTimeoutMs = "30000";
        this.keyDeserializer = "org.apache.kafka.common.serialization.StringDeserializer";
        this.valueDeserializer = "org.apache.kafka.common.serialization.StringDeserializer";
    }

    /**
     * SimpleKafkaConsumerFactory get() returns a org.apache.kafka.clients.consumer.KafkaConsumer.
     * @return KafkaConsumer
     */
    public <K, V> KafkaConsumer<K, V> build() {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", this.bootstrap);
        properties.put("group.id", this.groupId);
        properties.put("enable.auto.commit", this.enableAutoCommit);
        properties.put("auto.commit.interval.ms", this.autoCommitIntervalMs);
        properties.put("auto.offset.reset", this.autoOffsetReset);
        properties.put("session.timeout.ms", this.sessionTimeoutMs);
        properties.put("key.deserializer", this.keyDeserializer);
        properties.put("value.deserializer", this.valueDeserializer);
        return new KafkaConsumer<>(properties);
    }

    public SimpleKafkaConsumerBuilder setBootstrap(String bootstrap) {
        this.bootstrap = bootstrap;
        return this;
    }

    public SimpleKafkaConsumerBuilder setGroupId(String groupId) {
        this.groupId = groupId;
        return this;
    }

    public SimpleKafkaConsumerBuilder setEnableAutoCommit(String enableAutoCommit) {
        this.enableAutoCommit = enableAutoCommit;
        return this;
    }

    public SimpleKafkaConsumerBuilder setAutoCommitIntervalMs(String autoCommitIntervalMs) {
        this.autoCommitIntervalMs = autoCommitIntervalMs;
        return this;
    }

    public SimpleKafkaConsumerBuilder setAutoOffsetReset(String autoOffsetReset) {
        this.autoOffsetReset = autoOffsetReset;
        return this;
    }

    public SimpleKafkaConsumerBuilder setSessionTimeoutMs(String sessionTimeoutMs) {
        this.sessionTimeoutMs = sessionTimeoutMs;
        return this;
    }

    public SimpleKafkaConsumerBuilder setKeyDeserializer(String keyDeserializer) {
        this.keyDeserializer = keyDeserializer;
        return this;
    }

    public SimpleKafkaConsumerBuilder setValueDeserializer(String valueDeserializer) {
        this.valueDeserializer = valueDeserializer;
        return this;
    }
}
